package com.example.rocketmqdemo.message;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Delayed;

/**
 * @ClassName MyProducer1
 * @Description TODO
 * @Author 郭敖
 * @Date 2023年12月05日 22:56
 * @Version 1.0
 */
@Component
@Slf4j
public class MyProducer1 {
    @Autowired
    RocketMQTemplate rocketMQTemplate;

    public String sendMessage(String topic,String message){
        rocketMQTemplate.convertAndSend(topic, message);
        return "消息发送成功";
    }

    public void sendMessageAndCallback(String topic,String message){
        rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送消息的结果是：${}",sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                throwable.printStackTrace();
            }
        });
    }

    /*
    * 发送同步消息  又返回值  消息发送成功后才会结束
    * */
    public void sendMessageWithKey(String topic,String message){
        SendResult sendResult = rocketMQTemplate.syncSend(topic, message);
        log.info("同步消息的返回结果:${}",sendResult);
    }

    /**
     * 延时消息
     * @param topic
     * @param message
     */
    public void sendDelayedMessage(String topic,String message){
        Map<String,Object> map = new HashMap<String,Object>();
        map.put("key", UUID.randomUUID().toString());
        SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.createMessage(message, new MessageHeaders(map)), 3000, 4);
        log.info("同步延时消息为：${}",sendResult);
    }

    public void sendAsyncMessage(String topic,String message){
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(message)
                .setHeader("keys", UUID.randomUUID().toString()).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步延时消息的发送结果为： ${}",sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                throwable.printStackTrace();
            }
        },3000,4);
    }

    public void sendBroadcastMessage(String topic,String message){
        rocketMQTemplate.syncSend(topic,message);
    }

    /**
     *  发送同步事务消息
     * @param topic
     * @param message
     */
    @Transactional
    public void sendTransactionalMessage(String topic,String message){
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(topic, MessageBuilder.withPayload(message)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString()).build(), message);
        log.info("同步事务的消息的返回结果是:${}",transactionSendResult);
    }

    /**
     * 发送有序消息
     * @param topic 消息主题
     * @param message 消息内容
     */
    public void sendOrderlyMessage(String topic, String message) {
        // 发送有序消息 通过hashKey来决定发往同一topic中的哪个分区
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, MessageBuilder.
                withPayload(message).
                setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString()).
                setHeader(RocketMQHeaders.QUEUE_ID, 0).
                build(),"0");

        // 打印同步顺序消息的返回结果
        log.info("同步顺序的消息的返回结果是:{}", sendResult);
    }
}
